package cn.jcet.mapreduce.inputformat;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * @author 鲜磊
 * @Date 2020/12/8 11:28
 */
public class WholeRecordReader extends RecordReader<Text,BytesWritable>{

    // 文件切割
    FileSplit split;
    // 配置类
    Configuration configuration;
    // 设置一个标志位
    boolean flag = true;
    // 创建key和value
    Text k = new Text();
    BytesWritable v =  new BytesWritable();

    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {

         this.split = (FileSplit) inputSplit;
         configuration = taskAttemptContext.getConfiguration();
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (flag){
            FSDataInputStream fis = null;
            FileSystem fs = null;
            try {

                // 获取上传路径
                Path path = split.getPath();
                // 获取系统配置
                 fs = path.getFileSystem(configuration);
                // 获取输入流
                 fis = fs.open(path);
                // 创建缓冲区,整个文件大小
                byte [] buff = new byte[(int) split.getLength()];
                // 使用流拷贝,读取文件内容
                IOUtils.readFully(fis,buff,0,buff.length);
                // 输出整个文件内容
                v.set(buff,0,buff.length);
                // 获取文件路径以及名称
                String name = split.getPath().getName();
                // name作为key索引
                k.set(name);
            }catch (Exception e){

            }finally {
                IOUtils.closeStream(fis);
            }
            flag = false;
            return true;
        }
        return false;
    }

    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return k;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return v;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {

    }
}